package com.v.im.imservice.util;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class ZkWatchAPI implements Watcher {


    public static final Logger logger = LoggerFactory.getLogger(ZkWatchAPI.class);

    private static final int SESSION_TIMEOUT = 100010000;

    private ZooKeeper zk = null;

    private CountDownLatch connectedSemaphore = new CountDownLatch( 1 );

    private String monitorPath = "/";

    public ZkWatchAPI(String monitorPath) {
        this.monitorPath = monitorPath;
    }

    /**
     * 连接Zookeeper
     * @param connectString  Zookeeper服务地址
     */
    public void connectionZookeeper(String connectString){
        connectionZookeeper(connectString,SESSION_TIMEOUT);
    }

    /**
     * <p>连接Zookeeper</p>
     * <pre>
     *     [关于connectString服务器地址配置]
     *     格式: 192.168.1.1:2181,192.168.1.2:2181,192.168.1.3:2181
     *     这个地址配置有多个ip:port之间逗号分隔,底层操作
     *     ConnectStringParser connectStringParser =  new ConnectStringParser(“192.168.1.1:2181,192.168.1.2:2181,192.168.1.3:2181”);
     *     这个类主要就是解析传入地址列表字符串，将其它保存在一个ArrayList中
     *     ArrayList<InetSocketAddress> serverAddresses = new ArrayList<InetSocketAddress>();
     *     接下去，这个地址列表会被进一步封装成StaticHostProvider对象，并且在运行过程中，一直是这个对象来维护整个地址列表。
     *     ZK客户端将所有Server保存在一个List中，然后随机打乱(这个随机过程是一次性的)，并且形成一个环，具体使用的时候，从0号位开始一个一个使用。
     *     因此，Server地址能够重复配置，这样能够弥补客户端无法设置Server权重的缺陷，但是也会加大风险。
     *
     *     [客户端和服务端会话说明]
     *     ZooKeeper中，客户端和服务端建立连接后，会话随之建立，生成一个全局唯一的会话ID(Session ID)。
     *     服务器和客户端之间维持的是一个长连接，在SESSION_TIMEOUT时间内，服务器会确定客户端是否正常连接(客户端会定时向服务器发送heart_beat，服务器重置下次SESSION_TIMEOUT时间)。
     *     因此，在正常情况下，Session一直有效，并且ZK集群所有机器上都保存这个Session信息。
     *     在出现网络或其它问题情况下（例如客户端所连接的那台ZK机器挂了，或是其它原因的网络闪断）,客户端与当前连接的那台服务器之间连接断了,
     *     这个时候客户端会主动在地址列表（实例化ZK对象的时候传入构造方法的那个参数connectString）中选择新的地址进行连接。
     *
     *     [会话时间]
     *     客户端并不是可以随意设置这个会话超时时间，在ZK服务器端对会话超时时间是有限制的，主要是minSessionTimeout和maxSessionTimeout这两个参数设置的。
     *     如果客户端设置的超时时间不在这个范围，那么会被强制设置为最大或最小时间。 默认的Session超时时间是在2 * tickTime ~ 20 * tickTime
     * </pre>
     * @param connectString  Zookeeper服务地址
     * @param sessionTimeout Zookeeper连接超时时间
     */
    public void connectionZookeeper(String connectString, int sessionTimeout){
        this.releaseConnection();
        try {
            // ZK客户端允许我们将ZK服务器的所有地址都配置在这里
            zk = new ZooKeeper(connectString, sessionTimeout, this );
            // 使用CountDownLatch.await()的线程（当前线程）阻塞直到所有其它拥有CountDownLatch的线程执行完毕（countDown()结果为0）
            connectedSemaphore.await();
        } catch ( InterruptedException e ) {
            logger.error("连接创建失败，发生 InterruptedException , e " + e.getMessage(), e);
        } catch ( IOException e ) {
            logger.error( "连接创建失败，发生 IOException , e " + e.getMessage(), e );
        }
    }


    /**
     * Watcher Server,处理收到的变更
     * @param watchedEvent
     */
    @Override
    public void process(WatchedEvent watchedEvent) {
        if ( Event.KeeperState.SyncConnected == watchedEvent.getState() ) {
            connectedSemaphore.countDown();
            Event.EventType eventType = watchedEvent.getType();
            if (Event.EventType.None == eventType) {
                logger.debug("成功连接上ZK服务器 ");
                connectedSemaphore.countDown();
                try {
                    zk.exists(monitorPath, true);
                    List<String> paths = zk.getChildren(monitorPath, true);
                    for (String p : paths) {
                        this.zk.exists(monitorPath+ File.separator+p, true);
                    }
                } catch (KeeperException | InterruptedException e) {
                    e.printStackTrace();
                }
            }else{
                if (Event.EventType.NodeCreated == eventType) {
                    logger.info("节点创建 ");
                }else if (Event.EventType.NodeDataChanged == eventType) {
                    logger.info("节点数据更新");
                }else if (Event.EventType.NodeChildrenChanged == eventType) {
                    logger.info("子节点数据更新");
                }else if (Event.EventType.NodeDeleted == eventType) {
                    logger.info("节点删除");
                }
                try {
                    zk.exists(monitorPath, true);
                } catch (KeeperException | InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 关闭ZK连接
     */
    public void releaseConnection() {
        if ( null != zk ) {
            try {
                zk.close();
            } catch ( InterruptedException e ) {
                logger.error("release connection error ," + e.getMessage() ,e);
            }
        }
    }
}